/*
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.facebook.presto.adaptive;

import com.facebook.airlift.log.Logger;
import com.facebook.presto.Session;
import com.facebook.presto.adaptive.create.AdaptiveCreator;
import com.facebook.presto.adaptive.create.HintCreator;
import com.facebook.presto.adaptive.create.RedundantCreator;
import com.facebook.presto.metadata.Metadata;
import com.facebook.presto.metadata.QualifiedObjectName;
import com.facebook.presto.spi.ColumnHandle;
import com.facebook.presto.spi.TableHandle;
import com.facebook.presto.spi.adaptive.metadata.presto.AdaptivePrestoObject;
import com.facebook.presto.spi.adaptive.metadata.presto.PrestoMetadata;
import com.facebook.presto.spi.adaptive.metadata.presto.PrestoTable;
import com.facebook.presto.spi.session.PropertyMetadata;
import com.google.common.collect.ImmutableList;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;

import java.io.File;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

import static com.facebook.presto.spi.adaptive.AdaptiveConstant.TABLE_PROPERTY_CATALOG;
import static com.facebook.presto.spi.adaptive.AdaptiveConstant.TABLE_PROPERTY_ROWKEYFORMAT;
import static com.facebook.presto.spi.adaptive.AdaptiveConstant.TABLE_PROPERTY_MONGOINDEX;
import static com.facebook.presto.spi.session.PropertyMetadata.stringProperty;
import static com.facebook.presto.util.PropertiesUtil.loadProperties;

/**
 * @author luo
 * MetadataManager manages presto-metadata, hbase-metadata, neo4j-metadata.
 * All configs like ZK_HOST will read from adaptive.config file.
 * You can use getPrestoMetadataManager/getHBaseMetadataManager/getNeo4jMetaManager
 * and fill functions in these managers.
 */
public class AdaptiveMetadataManager
{
    public static final Logger log = Logger.get(AdaptiveMetadataManager.class);

    private File adaptiveConfigDir = new File("etc/adaptive/adaptive.config");
    private Map<String, String> properties = new HashMap<>();

    private String ZK_HOST;
    private String PRESTO_METADATA_PATH;

    private CuratorFramework client;
    private PrestoMetadata prestoMetadata;
    private List<AdaptivePrestoObject> adaptivePrestoObjectList = new ArrayList<>();

    private List<AdaptiveCreator> adaptiveCreators = ImmutableList.of(new HintCreator(), new RedundantCreator());

    private List<PropertyMetadata<?>> tableProperties = ImmutableList.of(
            stringProperty(
                    TABLE_PROPERTY_CATALOG,
                    "user hint of create table",
                    null,
                    false),
            stringProperty(
                    TABLE_PROPERTY_ROWKEYFORMAT,
                    "hint rowkey format",
                    null,
                    false),
            stringProperty(
                    TABLE_PROPERTY_MONGOINDEX,
                    "hint mongo index",
                    null,
                    false));

    /**
     * MetadataManager will connect to zk and all partial managers will get data from zkClient.
     */
    public AdaptiveMetadataManager()
    {
        loadConfig(adaptiveConfigDir);

        if (isAdaptEnable()) {
            CuratorFramework client = CuratorFrameworkFactory.builder()
                    .retryPolicy(new ExponentialBackoffRetry(1000, 3))
                    .connectString(ZK_HOST)
                    .build();
            client.start();

            this.client = client;
            this.prestoMetadata = readPrestoMetadataFromZK();
        }
    }

    /**
     * Load the config from external files.
     *
     * @param file the path of config file.
     */
    private void loadConfig(File file)
    {
        try {
            properties = loadProperties(file);
            ZK_HOST = properties.get("zk_host");
            PRESTO_METADATA_PATH = properties.get("presto_metadata_path");
        }
        catch (Exception ex) {
            log.error("Read adaptive config failed, " + ex.getMessage());
        }
    }

    /**
     * Read data and deserialize PrestoMetadata from zk.
     * If zkNode is null, it will return a prestoMetadata including empty prestoTable list.
     */
    private PrestoMetadata readPrestoMetadataFromZK()
    {
        try {
            return AdaptiveUtil.deserialize(client.getData().forPath(PRESTO_METADATA_PATH));
        }
        catch (Exception e) {
            log.error("Presto read metadata failed, check %s:%s", ZK_HOST, PRESTO_METADATA_PATH);
            return new PrestoMetadata();
        }
    }

    /**
     * Serialize PrestoMetadata and write data to zk.
     * If zkNode is null, it will create a node and write it.
     */
    private void writePrestoMetadataToZK(PrestoMetadata prestoMetadata)
    {
        try {
            if (client.checkExists().forPath(PRESTO_METADATA_PATH) == null) {
                client.create().creatingParentsIfNeeded().forPath(PRESTO_METADATA_PATH, AdaptiveUtil.serialize(prestoMetadata));
            }
            else {
                client.setData().forPath(PRESTO_METADATA_PATH, AdaptiveUtil.serialize(prestoMetadata));
            }
        }
        catch (Exception e) {
            log.error("Presto write metadata failed, check %s:%s", ZK_HOST, PRESTO_METADATA_PATH);
        }
    }

    /**
     * Return the state of adaptive module.
     *
     * @return true: adaptive module is on.
     * false: adaptive module is off.
     */
    public boolean isAdaptEnable()
    {
        String enable = properties.get("adaptive.enable");
        return enable != null && enable.equals("true");
    }

    /**
     * Get all adptive engines from config.
     */
    public List<String> getAdaptiveEngines()
    {
        String engines = properties.get("adaptive.engine");
        if (engines == null || engines.isEmpty()) {
            return ImmutableList.of();
        }
        return Arrays.asList(engines.split(","));
    }

    /**
     * Get adaptiveObjectMetadata(catalog/schema/table) by tableName
     * Init adapativeObjectList.
     */
    public void createAllAdaptiveObject(Session session, Metadata metadata)
    {
        try {
            for (PrestoTable prestoTable : getPrestoTableList()) {

                boolean tableExists = getAdaptivePrestoObjectList().stream()
                        .anyMatch(adaptivePrestoObject -> adaptivePrestoObject.getPrestoTable().equals(prestoTable));

                if (!tableExists) {
                    TableHandle tableHandle = metadata.getTableHandle(session,
                            new QualifiedObjectName(prestoTable.getCatalog(),
                                    prestoTable.getSchema(),
                                    prestoTable.getTable())).get();
                    Map<String, ColumnHandle> columnHandles = metadata.getColumnHandles(session, tableHandle);
                    adaptivePrestoObjectList.add(new AdaptivePrestoObject(prestoTable, tableHandle, columnHandles));
                    log.debug("Adaptive object register success. Adaptive name: %s", prestoTable);
                }
            }
        }
        catch (Exception ex) {
            adaptivePrestoObjectList.clear();
            log.error("Adaptive object register fail.");
        }
    }

    /**
     * Add a presto table.
     */
    public void addATable(String catalog, String schema, String table)
    {
        prestoMetadata.getPrestoTableList().add(new PrestoTable(catalog, schema, table));
        writePrestoMetadataToZK(prestoMetadata);
    }

    /**
     * Add a presto table.
     */
    public void addATable(PrestoTable prestoTable)
    {
        prestoMetadata.getPrestoTableList().add(prestoTable);
        writePrestoMetadataToZK(prestoMetadata);
    }

    /**
     * Remove a presto table.
     */
    public void removeATable(String catalog, String schema, String table)
    {
        prestoMetadata.getPrestoTableList().remove(new PrestoTable(catalog, schema, table));
        removeAPrestoObject(catalog, schema, table);
        writePrestoMetadataToZK(prestoMetadata);
    }

    /**
     * Remove a presto object.
     */
    private void removeAPrestoObject(String catalog, String schema, String table)
    {
        AdaptivePrestoObject prestoObject = getAdaptivePrestoObject(catalog, schema, table);
        if (prestoObject == null) {
            log.error("Remove presto object failed, prestoObjectList does not have the table, %s, %s %s.", catalog, schema, table);
            return;
        }
        getAdaptivePrestoObjectList().remove(prestoObject);
    }

    public List<PrestoTable> getPrestoTableList()
    {
        return prestoMetadata.getPrestoTableList();
    }

    public List<AdaptivePrestoObject> getAdaptivePrestoObjectList()
    {
        return adaptivePrestoObjectList;
    }

    /**
     * Get all table handles by table name.
     */
    public List<TableHandle> getTableHandlesByTableName(String tableName)
    {
        return
                getAdaptivePrestoObjectList().stream()
                        .filter(prestoObject -> prestoObject.getPrestoTable().getTable().equals(tableName))
                        .map(AdaptivePrestoObject::getTableHandle)
                        .collect(Collectors.toList());
    }

    /**
     * Get PrestoTable(catalog/schema/table) by tableName.
     * We record the table metadata when adaptive module is inited, and now return the first qualified object
     * (it may exist "tpcds.sf1.call_cetner" and "hbase.default.call_cener" because of redundancy storage)
     *
     * @param tableName table's name, such as "call_center".
     * @return adaptive name , such as "tpcds.sf1.call_center".
     */
    public PrestoTable getAdaptivePrestoTable(String tableName)
    {
        return prestoMetadata.getPrestoTableList().stream()
                .filter(prestoTable -> prestoTable.getTable().equals(tableName))
                .findFirst()
                .orElse(null);
    }

    /**
     * Check whether the specific catalog exists specific tableName.
     */
    public boolean existAdaptivePrestoTable(String catalogName, String tableName)
    {
        return prestoMetadata.getPrestoTableList().stream()
                .anyMatch(prestoTable -> prestoTable.equals(catalogName, "", tableName));
    }

    /**
     *  Check whether any catalog exists specific tableName.
     */
    public boolean existAdaptivePrestoTable(String tableName)
    {
        return prestoMetadata.getPrestoTableList().stream()
                .anyMatch(prestoTable -> prestoTable.getTable().equals(tableName));
    }

    /**
     * Get the adaptive object from adaptiveObjectList, by catalog, schema, table.
     */
    public AdaptivePrestoObject getAdaptivePrestoObject(String catalog, String schema, String table)
    {
        return adaptivePrestoObjectList.stream()
                .filter(adaptiveObject -> adaptiveObject.equals(catalog, schema, table))
                .findFirst()
                .orElse(null);
    }

    /**
     * Get adaptive presto tables by table name.
     */
    public List<PrestoTable> getPrestoTablesByTableName(String tableName)
    {
        return prestoMetadata.getPrestoTableList().stream()
                .filter(prestoTable -> prestoTable.getTable().equals(tableName))
                .collect(Collectors.toList());
    }

    /**
     * Get catalogs list by table name.
     */
    public List<String> getCatalogListByTableName(String tableName)
    {
        return prestoMetadata.getPrestoTableList().stream()
                .filter(prestoTable -> prestoTable.getTable().equals(tableName))
                .map(PrestoTable::getCatalog)
                .collect(Collectors.toList());
    }

    /**
     * Get adaptive creator rules.
     */
    public List<AdaptiveCreator> getAdaptiveCreators()
    {
        return adaptiveCreators;
    }

    /**
     * Get default schema name for each catalog.
     */
    public String getDefaultSchema(String catalog)
    {
        return "default";
    }

    public List<PropertyMetadata<?>> getTableProperties()
    {
        return tableProperties;
    }
}
